package org.dromara.common.sse.listener;

import cn.hutool.core.collection.CollUtil;
import com.cloud.rocketmq.base.entity.CloudMessage;
import com.cloud.rocketmq.spring.annotation.ConsumeTopic;
import com.cloud.rocketmq.spring.core.TopicListener;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.sse.core.SseEmitterManager;
import org.dromara.common.sse.dto.SseMessageDto;

import static org.dromara.common.core.constant.RocketMQConstants.SseTopic.SSE_MSG_TAG_NAME;
import static org.dromara.common.core.constant.RocketMQConstants.SseTopic.SSE_MSG_TOPIC_NAME;

/**
 * SSE 主题订阅监听器
 *
 * @author shuai.zhou
 */
@Slf4j
@ConsumeTopic(topic = SSE_MSG_TOPIC_NAME, eventCode = SSE_MSG_TAG_NAME, log = true)
@RequiredArgsConstructor
public class SseMsgMQListener implements TopicListener<SseMessageDto> {

    private final SseEmitterManager sseEmitterManager;

    @Override
    public void onMessage(CloudMessage<SseMessageDto> cloudMessage) {
        SseMessageDto message = cloudMessage.getPayload();
        // 如果key不为空就按照key发消息 如果为空就群发
        if (CollUtil.isNotEmpty(message.getUserIds())) {
            message.getUserIds().forEach(key -> {
                sseEmitterManager.sendMessage(key, message.getMessage());
            });
        } else {
            sseEmitterManager.sendMessage(message.getMessage());
        }
    }

}
